package com.mikasa.rabbitmq.api.dlx;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import com.rabbitmq.client.*;

/**
 * @author mikasa
 */
public class Consumer {

	
	public static void main(String[] args) throws Exception {
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost("39.108.108.234");
		connectionFactory.setPort(5672);
		connectionFactory.setVirtualHost("/");

		Connection connection = connectionFactory.newConnection();
		Channel channel = connection.createChannel();


		// 普通交换机、队列名
		String exchangeName = "test_dlx_exchange";
		String queueName = "test_dlx_que";
		String routingKey = "dlx.#";

		// 声明普通交换机
		channel.exchangeDeclare(exchangeName, "topic", true, false, null);

		Map<String, Object> arg = new HashMap<>();
		// 指定死信发送的交换机
		arg.put("x-dead-letter-exchange", "dlx_exchange");
		// 声明普通队列
		// 这个agr要声明在队列上，表示这个队列的死信会被发送到哪个指定的交换机上
		channel.queueDeclare(queueName, true, false, false,arg);

		// 绑定交换机和队列
		channel.queueBind(queueName, exchangeName, routingKey);

		// 声明死信队列
		String dlxExchangeName = "dlx_exchange";
		String dlxQueueName = "dlx_queue";
		// 发送给任意所有绑定交换机的队列
		String dlxRoutingKey = "#";
		channel.exchangeDeclare(dlxExchangeName, "topic", true, false, null);
		channel.queueDeclare(dlxQueueName, true, false, false, null);
		channel.queueBind(dlxQueueName, dlxExchangeName, dlxRoutingKey);

		// 接收普通消息
		channel.basicConsume(queueName, false, new DefaultConsumer(channel){
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
				System.out.println("处理消息");
				System.err.println("-----------consume message----------");
				System.err.println("consumerTag: " + consumerTag);
				System.err.println("envelope: " + envelope);
				System.err.println("properties: " + properties);
				System.err.println("body: " + new String(body));
				// autoAck为false且没有手动接收消息，就会被uack变成死信消息
				// channel.basicAck(envelope.getDeliveryTag(), false);
			}
		});

		// 接收死信消息
		channel.basicConsume(dlxQueueName, false, new DefaultConsumer(channel){
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException{
				System.out.println("处理死信消息");
				System.err.println("-----------consume message----------");
				System.err.println("consumerTag: " + consumerTag);
				System.err.println("envelope: " + envelope);
				System.err.println("properties: " + properties);
				System.err.println("body: " + new String(body));
				// ack
				channel.basicAck(envelope.getDeliveryTag(), false);
			}
		});
	}
		
}


